package com.k2data.hbase;

import com.sun.org.apache.commons.logging.LogFactory;
import com.sun.org.apache.commons.logging.Log;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Created by luoqifei on 17-2-20.
 */
public class ObserverCoprocessor extends BaseRegionObserver{
    public static final Log LOG = LogFactory.getLog(ObserverCoprocessor.class);
    private String splitStr = "_";
    private String pStatusC = "pStatus";
    private String finishedStatus = "finished";
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        RegionCoprocessorEnvironment env = e.getEnvironment();
        String thisTableName = Bytes.toString(env.getRegion().getTableDesc().getTableName().getName());
        LOG.info("coprocessor TableName is : "+thisTableName);
        Table updateTable = env.getTable(TableName.valueOf(thisTableName));
        try{
            //get the post put row content
            byte[] rowKey = put.getRow();
            LOG.info("put record old-rowKey is "+new String(rowKey));
            String oldRowKey = new String(rowKey);
            //check if it was update record
            if(oldRowKey.indexOf(splitStr) == oldRowKey.lastIndexOf(splitStr)){
                return;
            }
            //old row key is pName_pType_runDate,new key is  pName_pType
            String newRowKey = oldRowKey.substring(0,oldRowKey.lastIndexOf(splitStr));
            Put updatePut = new Put(newRowKey.getBytes());
            CellScanner scanner = put.cellScanner();
            while (scanner.advance()){
                Cell cell = scanner.current();
                long timestamp = cell.getTimestamp();
                //LOG.debug("timestamp is "+timestamp);
                byte[] columnFamily = CellUtil.cloneFamily(cell);
                //LOG.debug("columnFamily is "+new String(columnFamily));
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                LOG.debug("qualifier is "+new String(qualifier));
                byte[] value = CellUtil.cloneValue(cell);
                LOG.debug("value is "+new String(value));
                if(((new String(qualifier)).contains(pStatusC))){
                    if(!(new String(value)).equals(finishedStatus)){
                        LOG.info("wt status isn't finished, coprocessor do not deal this Put.");
                        return;
                    }else {
                        LOG.info("Great! wt status is finished.");
                    }
                }
                updatePut.addColumn(columnFamily,qualifier,timestamp,value);
            }
            LOG.info("start to insert a new record and it's newRowKey is "+newRowKey);
            updateTable.put(updatePut);
        }finally {
            updateTable.close();
        }
    }
}
